package com.ctsi.canal.starter.config.configuration;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.ctsi.canal.starter.client.ClusterCanalClient;
import com.ctsi.canal.starter.config.properties.CanalProperties;
import com.ctsi.canal.starter.config.properties.SimpleCanalProperties;
import com.ctsi.canal.starter.factory.EntryColumnModelFactory;
import com.ctsi.canal.starter.handler.EntryHandler;
import com.ctsi.canal.starter.handler.MessageHandler;
import com.ctsi.canal.starter.handler.RowDataHandler;
import com.ctsi.canal.starter.handler.impl.AsyncMessageHandlerImpl;
import com.ctsi.canal.starter.handler.impl.RowDataHandlerImpl;
import com.ctsi.canal.starter.handler.impl.SyncMessageHandlerImpl;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * canal 集群客户端配置
 *
 * @author ajh
 * @date 2019/10/24 15:22
 **/
@Configuration
@EnableConfigurationProperties(SimpleCanalProperties.class)
@ConditionalOnBean(value = { EntryHandler.class })
@ConditionalOnProperty(value = CanalProperties.CANAL_MODE, havingValue = "cluster")
@Import(ThreadPoolConfiguration.class)
public class ClusterClientConfiguration {

    private final SimpleCanalProperties simpleCanalProperties;

    public ClusterClientConfiguration(SimpleCanalProperties simpleCanalProperties) {
        this.simpleCanalProperties = simpleCanalProperties;
    }

    @Bean
    public RowDataHandler<CanalEntry.RowData> rowDataHandler() {
        return new RowDataHandlerImpl(new EntryColumnModelFactory());
    }

    @Bean
    @ConditionalOnProperty(value = CanalProperties.CANAL_ASYNC, havingValue = "true", matchIfMissing = true)
    public MessageHandler messageHandler(RowDataHandler<CanalEntry.RowData> rowDataHandler,
                                         List<EntryHandler> entryHandlers,
                                         ExecutorService executorService) {
        return new AsyncMessageHandlerImpl(entryHandlers, rowDataHandler, executorService);
    }

    @Bean
    @ConditionalOnProperty(value = CanalProperties.CANAL_ASYNC, havingValue = "false")
    public MessageHandler messageHandler(RowDataHandler<CanalEntry.RowData> rowDataHandler,
                                         List<EntryHandler> entryHandlers) {
        return new SyncMessageHandlerImpl(entryHandlers, rowDataHandler);
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    public ClusterCanalClient clusterCanalClient(MessageHandler messageHandler) {
        return ClusterCanalClient.Builder.builder()
                .canalServers(simpleCanalProperties.getServer())
                .destination(simpleCanalProperties.getDestination())
                .userName(simpleCanalProperties.getUsername())
                .messageHandler(messageHandler)
                .password(simpleCanalProperties.getPassword())
                .batchSize(simpleCanalProperties.getBatchSize())
                .filter(simpleCanalProperties.getFilter())
                .timeout(simpleCanalProperties.getTimeout())
                .unit(simpleCanalProperties.getUnit()).build();
    }

}
